package cn.xinfei.xdecision.pulsar.consumer;

import cn.xinfei.xdecision.common.utils.util.JsonUtils;
import cn.xinfei.xdecision.pulsar.model.XPulsarConsumerModel;
import com.alibaba.fastjson.JSONObject;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;

import java.util.concurrent.CompletableFuture;

@Data
@Slf4j
public class XPulsarConsumer<T> {
    private Consumer<T> consumer;
    private XPulsarConsumerModel xPulsarConsumerModel;

    public XPulsarConsumer(Consumer<T> consumer, XPulsarConsumerModel xPulsarConsumerModel) {
        this.consumer = consumer;
        this.xPulsarConsumerModel = xPulsarConsumerModel;
    }

    public void stop() throws PulsarClientException {
        if (consumer.isConnected()) {
            log.warn("pulsar consumer is closing .topic={},consumerName={}", consumer.getTopic(), consumer.getConsumerName());
            ConsumerStats stats = consumer.getStats();
            log.info("pulsar consumer stats={}", JsonUtils.toJsonString(stats));
            consumer.close();
        }
        if (!consumer.isConnected()) {
            log.warn("pulsar consumer is closed .topic={},consumerName={}", consumer.getTopic(), consumer.getConsumerName());
        }
    }

    public Consumer<T> getConsumer() {
        return consumer;
    }

    public CompletableFuture<Messages<T>> batchReceiveAsync() {
        return consumer.batchReceiveAsync();
    }

    public CompletableFuture<Message<T>> receiveAsync() {
        return consumer.receiveAsync();
    }

    public Messages<T> batchReceive() throws PulsarClientException {
        return consumer.batchReceive();
    }

    public String getTopic() {
        return consumer.getTopic();
    }

    public ConsumerStats getStats() {
        return consumer.getStats();
    }


    public void close() throws PulsarClientException {
        consumer.close();
    }


    public CompletableFuture<Void> closeAsync() {
        return consumer.closeAsync();
    }


    public boolean isConnected() {
        return consumer.isConnected();
    }


    public String getConsumerName() {
        return consumer.getConsumerName();
    }
}
